package com.qubi.exchange.websocket;


import com.alibaba.fastjson.JSONObject;
import com.qubi.exchange.message.NotifyMessage;
import com.qubi.exchange.queue.DelayMessage;
import com.qubi.exchange.queue.DelayQueueBySpeed;
import com.qubi.exchange.util.GZipUtils;
import lombok.Data;
import lombok.extern.log4j.Log4j2;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
import org.springframework.stereotype.Service;

import javax.websocket.Session;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Log4j2
public class NotifyMessageContainer  implements ApplicationRunner, Ordered {
    private static int delay=250;
     static Map<Session,SessonMessageContainer>blockingQueueMap=new ConcurrentHashMap<>(2000) ;

    private  static DelayQueue<DelayMessage<SessonMessageContainer>> queue= new DelayQueue<>();

    public static ExecutorService executor=Executors.newFixedThreadPool(20);

    public void putContainer( SessonMessageContainer container)
    {
        blockingQueueMap.put(container.getSession(),container);
    }


    public void remove( Session session ){

        blockingQueueMap.remove(session);

    }
    public void putMessage(Session session,NotifyMessage notifyMessage)
    {

        var container= blockingQueueMap.getOrDefault(session,new SessonMessageContainer(session) );
        container.putMessage(notifyMessage);
        DelayMessage delayMessage=new DelayMessage(delay,TimeUnit.MILLISECONDS,container);
        if(container.switchStatus()) {
            queue.offer(delayMessage);
        }
    }


    public void hanleMessage(Session session,NotifyMessage[]  messages)
    {
        log.info("发送数据: "+ JSONObject.toJSONString(messages));
        log.info("发送数据主题数: "+ messages.length);

        byte[] data  = JSONObject.toJSONBytes(messages);
        log.info("压缩前长度: "+ data.length);

        if(data.length>180) {
            try {
                data = GZipUtils.gzip(data);
                log.info("压缩后长度: "+ data.length);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        if (session != null) {

                if (session.isOpen()) {
                    synchronized (session) {
                    session.getAsyncRemote().sendBinary(ByteBuffer.wrap(data));
                }
            }
        }
    }

    ExecutorService service=Executors.newSingleThreadScheduledExecutor();

    @Override
    public void run(ApplicationArguments args) {
        service.submit(()->{
            while(true){
                try {
                    var delay=queue.take();
                    executor.submit(()->{
                        hanleMessage(delay.getData().getSession(),delay.getData().getAll());
                    });
                }catch (Exception e){
                    log.error(e.toString());
                }
            }

        });

    }

    @Override
    public int getOrder() {
        return 0;
    }

    @Data
    public  static class SessonMessageContainer {
        Session session;
        List<NotifyMessage> blockingQueue=new LinkedList<>() ;


        AtomicBoolean atomicBoolean=new AtomicBoolean(true);


        public boolean switchStatus()
        {
            return atomicBoolean.compareAndSet(true,false);
        }


        SessonMessageContainer(Session session){
            this.session=session;
        }
        public void putMessage(NotifyMessage notifyMessage)
        {
            synchronized (this) {
                blockingQueue.add(notifyMessage);
            }
        }
        public NotifyMessage[] getAll()
        {
            synchronized (this) {

                    var data = (NotifyMessage[]) blockingQueue.toArray(new NotifyMessage[0] );
                    blockingQueue.clear();
                    atomicBoolean.compareAndSet(false,true);

                    return data;

            }
        }
    }





}
